package com.intretech.tj.mqtt.Source;

import com.intretech.tj.global.parent.TriggerParent;
import com.intretech.tj.mqtt.OutLog;
import com.intretech.tj.mqtt.entity.MessageInfo;
import com.intretech.tj.mqtt.entity.QosMessageInfo;
import com.intretech.tj.mqtt.entity.RetainMessageInfo;
import com.intretech.tj.mqtt.entity.Subscriber;
import com.intretech.tj.mqtt.trigger.SubscribeTrigger;
import com.intretech.tj.mqtt.trigger.TopicNewTrigger;
import io.netty.handler.codec.mqtt.MqttProperties;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.smallrye.mutiny.Uni;
import io.vertx.mqtt.messages.codes.MqttDisconnectReasonCode;
import io.vertx.mutiny.core.Vertx;
import io.vertx.mutiny.core.buffer.Buffer;
import io.vertx.mutiny.mqtt.MqttEndpoint;
import io.vertx.mutiny.mqtt.messages.MqttPublishMessage;
import jakarta.inject.Inject;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.stream.Collectors;

import static com.intretech.tj.mqtt.OutLog.*;

@Slf4j
public class Source implements EndpointInterface, TopicInterface, MessageInterface {
    @Inject
    protected OutLog outLog;

    @Inject
    protected SubscribeTrigger subscribeTrigger;

    @Inject
    protected TopicNewTrigger topicNewTrigger;

    @Override
    public List<String> getAllClient() {
        return endpointMap.keySet().stream().toList();
    }

    @Override
    public void addEndpoint(MqttEndpoint newEndpoint) {
        String key = newEndpoint.clientIdentifier();
        if (endpointMap.containsKey(key))
        // 会话接管结束
        {
            endpointMap.get(key).disconnect(MqttDisconnectReasonCode.SESSION_TAKEN_OVER, new MqttProperties());
            // v5会通往close处理器进行删除，但是为了兼容v3，后面多进行一次删除
            if (endpointMap.get(key).isConnected())
                endpointMap.get(key).close();
        }
        endpointMap.put(newEndpoint.clientIdentifier(), newEndpoint);
    }

    @Override
    public void removeEndpoint(String clientID) {
        endpointMap.remove(clientID);
    }

    @Override
    public void change(String topic, boolean addKey) {
        if (addKey)
            topicMap.put(topic, new CopyOnWriteArrayList<>());
        else
            topicMap.remove(topic);
        topicNewTrigger.trigger(TriggerParent.ALL, topic);
    }

    @Override
    public void messageAdd(String topic) {
        if (!topicMap.containsKey(topic))
            change(topic, true);
    }

    @Override
    public void subscribe(String topic, MqttQoS qos, String clientID) {
        List<Subscriber> infos = topicMap.get(topic);
        subscribeTrigger.trigger(topic, clientID);
        if (Objects.isNull(infos)) {
            change(topic, true);
        } else
            for (Subscriber subscriber : topicMap.get(topic)) {
                if (subscriber.getClientID().equals(clientID)) {
                    subscriber.setQos(qos);
                    return;
                }
            }
        topicMap.get(topic).add(new Subscriber(qos, clientID));
    }

    @Override
    public void unsubscribe(String topic, String clientID) {
        topicMap.get(topic).removeIf(subscriber -> subscriber.getClientID().equals(clientID));
        if (topicMap.get(topic).isEmpty()) {
            change(topic, false);
        }
    }

    @Override
    public void unsubscribe(String clientID) {
        for (Map.Entry<String, List<Subscriber>> item : topicMap.entrySet()) {
            item.getValue().removeIf(subscriber -> subscriber.getClientID().equals(clientID));
        }
    }

    @Override
    public Map<String, List<Subscriber>> getTopicsForWildcard(String topic) {
        List<String> topics = new ArrayList<>();
        String[] topicSplit = topic.split("/");
        topicMap.keySet().forEach(key -> {
            if (key.equals(topic)) {
                topics.add(key);
            } else {
                String[] keySplit = key.split("/");
                int i = -1, j = -1;
                while (true) {
                    i++;
                    j++;
                    if (i == topicSplit.length || j == keySplit.length) {
                        if (topicSplit.length == keySplit.length) topics.add(key);
                        break;
                    }
                    if (topicSplit[i].equals("#") || keySplit[j].equals("#")) {
                        topics.add(key);
                        break;
                    }
                    if (topicSplit[i].equals("+") || keySplit[j].equals("+")) continue;
                    if (!topicSplit[i].equals(keySplit[j])) break;
                }

            }
        });
        return topics.stream()
                .filter(result->!topicMap.get(result).isEmpty())
                .collect(Collectors.toMap(result->result, topicMap::get));
    }

    @Override
    public List<String> getTopicList() {
        return topicMap.keySet().stream().toList();
    }

    @Override
    public void receive(MessageInfo messageInfo) {
        if (messageInfo.getMessage().isRetain())
            retainMessageMap.put(messageInfo.getMessage().topicName(), new RetainMessageInfo(messageInfo.getMessage().qosLevel(), messageInfo.getMessage().payload()));
        if (messageInfo.getMessage().isDup())
            qosMessageMap.remove(messageInfo.getMessage().messageId());
        if (messageInfo.getMessage().qosLevel() != MqttQoS.AT_MOST_ONCE)
            qosMessageMap.put(messageInfo.getMessage().messageId(), new QosMessageInfo(messageInfo));
    }

    /**
     * 从各种回复中判断是否完成一次完整的发布<br/>
     * 若订阅者全都完成接收，则回复发布者<br/>
     * 最后删除消息记录
     *
     * @param clientId  客户端id
     * @param messageId 消息id
     */
    protected void reply(String clientId, Integer messageId) {
        if (qosMessageMap.containsKey(messageId)) {
            QosMessageInfo qosMessageInfo = qosMessageMap.get(messageId);
            if (!Objects.isNull(clientId)) {
                qosMessageInfo.getCount().getAndDecrement();
                qosMessageInfo.getSucceedList().add(clientId);
            }
            // 判断订阅者是否全部收到
            if (qosMessageInfo.getCount().get() <= 0) {
                // 向发布者回复
                if (qosMessageInfo.getMessage().qosLevel().equals(MqttQoS.AT_LEAST_ONCE))
                    endpointMap.get(qosMessageInfo.getClientID()).publishAcknowledge(qosMessageInfo.getMessage().messageId());
                else if (qosMessageInfo.getMessage().qosLevel().equals(MqttQoS.EXACTLY_ONCE))
                    endpointMap.get(qosMessageInfo.getClientID()).publishComplete(qosMessageInfo.getMessage().messageId());
                // 删除消息记录
                qosMessageMap.remove(messageId);
            }
        }
    }

    /**
     * 向订阅者转发消息
     *
     * @param message 消息
     */
    protected void transmit(MqttPublishMessage message) {
        messageAdd(message.topicName());
        Map<String, List<Subscriber>> topicsMap = getTopicsForWildcard(message.topicName());
        // 检查是否存在一个有效的topic
        if (topicsMap.isEmpty()) {
            reply(null, message.messageId());
            return;
        }
        // 封装函数列表
        List<Callable<Subscriber>> runnableList = new CopyOnWriteArrayList<>();
        // 处理发布者qos=0
        if (message.qosLevel().equals(MqttQoS.AT_MOST_ONCE)) {
            topicsMap.forEach((topic, subscribers) ->
                    subscribers.forEach(subscriber -> runnableList.add(() -> {
                                publish(subscriber.getClientID(), topic, MqttQoS.AT_MOST_ONCE, message);
                                return subscriber;
                            })
                    )
            );
        } else {
            QosMessageInfo qosMessageInfo = qosMessageMap.get(message.messageId());
            topicsMap.forEach((topic, subscribers) -> subscribers.forEach(subscriber -> runnableList.add(() -> {
                        MqttQoS transmitQos = getQos(subscriber.getQos(), message.qosLevel());
                        qosMessageInfo.getCount().incrementAndGet();
                        // qos=0直接调用一次回复检查
                        if (transmitQos.equals(MqttQoS.AT_MOST_ONCE))
                            reply(subscriber.getClientID(), message.messageId());
                        publish(subscriber.getClientID(), topic, transmitQos, message);
                        return subscriber;
                    })
            ));
        }

        // 异步执行
        runnableList.forEach(runnable ->
                Vertx.currentContext().executeBlocking(runnable).subscribe().with(subscriber ->
                        log.debug(outLog.getMessage(Transmit, message.messageId(), subscriber.getClientID(), subscriber.getQos().value(), message.payload().toString())))
        );
    }

    /**
     * 获取qos等级</br>
     * qos降级</br>
     * 主题qos与消息qos对比，谁小选谁
     *
     * @param s1 主题qos
     * @param s2 消息qos
     * @return qos
     */
    protected MqttQoS getQos(MqttQoS s1, MqttQoS s2) {
        return s1.value() < s2.value() ? s1 : s2;
    }

    /**
     * 服务发布
     *
     * @param clientID 客户端id
     * @param topic    订阅的主题
     * @param qos      qos等级
     * @param message  原始消息
     */
    private void publish(String clientID, String topic, MqttQoS qos, MqttPublishMessage message) {
        Uni<Integer> uni;
        if (qos.equals(MqttQoS.AT_MOST_ONCE))
            uni = endpointMap.get(clientID).publish(topic, message.payload(), qos, message.isDup(), message.isRetain());
        else
            uni = endpointMap.get(clientID).publish(topic, message.payload(), qos, message.isDup(), message.isRetain(), message.messageId());
        uni.subscribe().with(id ->
                log.debug(outLog.getMessage(Publish, id, clientID, qos.value(), message.payload().toString()))
        );
    }

    /**
     * 调用发布
     *
     * @param clientID 客户端id
     * @param qosValue qos等级
     * @param topic    订阅的主题
     * @param message  原始消息
     */
    public void publish(String clientID, int qosValue, String topic, String message) {
        messageAdd(topic);
        endpointMap.get(clientID).publish(topic, Buffer.buffer(message.getBytes()), MqttQoS.valueOf(qosValue), false, false)
                .subscribe()
                .with(id -> log.debug(outLog.getMessage(Publish, id, clientID, MqttQoS.valueOf(qosValue), message)));
    }

    /**
     * 保留消息发布
     *
     * @param clientID 接收者id
     * @param qos      qos等级
     * @param topic    主题
     * @param buffer   字节信息
     */
    public void retain(String clientID, MqttQoS qos, String topic, Buffer buffer) {
        messageAdd(topic);
        endpointMap.get(clientID).publish(topic, buffer, qos, false, true)
                .subscribe()
                .with(id -> log.debug(outLog.getMessage(Retain, id, clientID, qos.value(), buffer.toString())));
    }

    /**
     * 调用写入保留消息
     *
     * @param topic   主题
     * @param qos     qos等级
     * @param message 消息
     */
    public void setRetain(String topic, MqttQoS qos, String message) {
        retainMessageMap.put(topic, new RetainMessageInfo(qos, Buffer.buffer(message)));
    }
}
